package net.gd.servlet;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.apache.log4j.Logger;
import net.gd.dao.UdfInfoDao;
import net.gd.dao.UdfModelInfoDao;
import net.gd.domain.UdfInfo;
import net.gd.domain.UdfModelLog;
import net.gd.utils.ZeppelinApiUtils;
import net.sf.json.JSONObject;

public class UdfInfoServlet extends BaseServlet {

  private static Logger logger = Logger.getLogger(UdfInfoServlet.class);

  /**
   * 
   */
  private static final long serialVersionUID = -1686069636428282319L;

  private UdfInfoDao udfInfoDao;
  private UdfModelInfoDao modelInfoDao;


  @Override
  protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws UnsupportedEncodingException, IOException {
    String path = req.getRequestURI();

      logger.info("uri: " + path);
      BufferedReader br = new BufferedReader(new InputStreamReader(req.getInputStream(), "UTF-8"));
      StringBuilder jsonInput = new StringBuilder();
      String line = null;
      while ((line = br.readLine()) != null) {
        jsonInput.append(line);
      }
      br.close();
      logger.info("input: " + jsonInput.toString() + " len:" + jsonInput.length());

      if (path.endsWith("save")) {
        // 新建或更新UDF
        save(req, resp, jsonInput.toString());
      } else if (path.endsWith("register")) {
        // 向zeppelin注册UDF
        register(req, resp, jsonInput.toString());
      } else if (path.endsWith("search")) {
        // 查询UDF信息
        search(req, resp, jsonInput.toString());
      }
    
    logger.info(path + " 请求结束。");
  }

  /**
   * 查询UDF信息
   * 
   * @param req
   * @param resp
   * @param input
   */
  private void search(HttpServletRequest req, HttpServletResponse resp, String input) {
    Map<String, Object> result = new HashMap<String, Object>();
    try {
      String udfId = null;
      String udfName = null;

      if (input.length() != 0) {
        JSONObject json = JSONObject.fromObject(input);
        if (json.containsKey("udf_id")) {
          udfId = json.getString("udf_id");
        } else if (json.containsKey("udf_name")) {
          udfName = json.getString("udf_name");
        }
      }
      List<UdfInfo> list = udfInfoDao.list(udfId, udfName);
      List<JSONObject> listJson = new ArrayList<JSONObject>();
      for (UdfInfo item : list) {
        listJson.add(item.toJson());
      }

      result.put("code", "0");
      result.put("list", listJson);
      saveText4Ajax(resp, JSONObject.fromObject(result).toString());
      return;
    } catch (Exception e) {
      logger.warn("Error: ", e);
      result.put("code", "1");
      result.put("msg", e.getMessage());
    }
    try {
      saveText4Ajax(resp, JSONObject.fromObject(result).toString());
    } catch (IOException e) {
      logger.error("response error.", e);
    }
  }


  /**
   * 向zeppelin注册UDF
   * 
   * @param req
   * @param resp
   * @param input
   */
  private void register(HttpServletRequest req, HttpServletResponse resp, String input) {
    Map<String, Object> result = new HashMap<String, Object>();
    try {

      boolean isError = false;
      String msg = "";
      JSONObject json = JSONObject.fromObject(input);

      if (!json.containsKey("udf_id")) {
        isError = true;
        msg = "udf_id is empty.";
      }
      UdfInfo udfInfo = udfInfoDao.get(json.getString("udf_id"));
      if (udfInfo == null) {
        isError = true;
        msg = "udf info is not exists.";
      }

      if (isError) {
        result.put("code", "1");
        result.put("msg", msg);
        saveText4Ajax(resp, JSONObject.fromObject(result).toString());
        return;
      }

      String noteId = json.getString("note_id");
      logger.info("param note_id = " + noteId);
      if(noteId == null || noteId.trim().length() == 0) {
        noteId = ZeppelinApiUtils.createNote();  
      }
      
      String paragraphs = ZeppelinApiUtils.PREFIX_SCALA + "\n";
      paragraphs += "sqlContext.udf.register(\"" + udfInfo.getUdfName() + "\", ";
      paragraphs += udfInfo.getUdfContent() + ")";
      String paragraphId = ZeppelinApiUtils.createParagraphs(noteId, paragraphs);
      Map<String, Object> rs = ZeppelinApiUtils.runParagraph(noteId, paragraphId);
      logger.info("register result: {}" + rs);
      if("OK".equals(rs.get("status").toString())) {
        result.put("code", "0");  
        result.put("note_id", noteId);
      } else {
        result.put("code", "1");
        result.put("msg", rs.get("body").toString());
      }
      
      saveText4Ajax(resp, JSONObject.fromObject(result).toString());
      return;
    } catch (Exception e) {
      logger.warn("Error: ", e);
      result.put("code", "1");
      result.put("msg", e.getMessage());
    }
    try {
      saveText4Ajax(resp, JSONObject.fromObject(result).toString());
    } catch (IOException e) {
      logger.error("response error.", e);
    }
  }

  private void save(HttpServletRequest req, HttpServletResponse resp, String input) {
    Map<String, Object> result = new HashMap<String, Object>();
    try {
      boolean isError = false;
      String msg = "";
      JSONObject json = JSONObject.fromObject(input);
      UdfInfo info = new UdfInfo();
      if (json.containsKey("udf_id")) {
        info.setUdfId(json.getString("udf_id"));
        UdfInfo oldInfo = udfInfoDao.get(info.getUdfId());
        if (oldInfo == null) {
          isError = true;
          msg = info.getUdfId() + " UdfInfo not exists";
        }
      }
      if (json.containsKey("udf_name")) {
        info.setUdfName(json.getString("udf_name"));
      }
      if (json.containsKey("udf_content")) {
        info.setUdfContent(json.getString("udf_content"));
      } else {
        isError = true;
        msg = "model_content is empty.";
      }
      if (json.containsKey("create_user")) {
        info.setCreateUser(json.getString("create_user"));
      }

      if (isError) {
        result.put("code", "1");
        result.put("msg", msg);
        saveText4Ajax(resp, JSONObject.fromObject(result).toString());
        return;
      }

      info.setLangType("scala");
      String id = udfInfoDao.saveOrUpdate(info);
      result.put("code", "0");
      result.put("udf_id", id);
      logger.info("save success");

      String testContent = "sqlContext.udf.register(\"" + info.getUdfName() + "\", ";
      testContent += info.getUdfContent() + ")";
      UdfModelLog log = ZeppelinApiUtils.runModelContent(info.getLangType(), testContent);
      logger.info("test result: " + log);
      if("SUCCESS".equals(log.getOperationStatus())) {
        result.put("code", "0");
      } else {
        result.put("code", "1");
        result.put("msg", log.getOperationResult());
      }

      saveText4Ajax(resp, JSONObject.fromObject(result).toString());
      return;
    } catch (Exception e) {
      result.put("code", "1");
      result.put("msg", e.getMessage());
      logger.warn("Error: ", e);
    }

    try {
      saveText4Ajax(resp, JSONObject.fromObject(result).toString());
    } catch (IOException e) {
      logger.error("response error.", e);
    }
  }

  public UdfInfoDao getUdfInfoDao() {
    return udfInfoDao;
  }

  public void setUdfInfoDao(UdfInfoDao udfInfoDao) {
    this.udfInfoDao = udfInfoDao;
  }

  public UdfModelInfoDao getModelInfoDao() {
    return modelInfoDao;
  }

  public void setModelInfoDao(UdfModelInfoDao modelInfoDao) {
    this.modelInfoDao = modelInfoDao;
  }



}
